package com.shujia.spark.core

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

object Demo13ReduceByKey {
  def main(args: Array[String]): Unit = {

    /**
     * reduceByKey：转换算子，需要作用在KV格式的RDD上
     * 需要接收一个聚合的函数f
     * 不仅会对Key进行分组，还会对V进行聚合，同时还会做预聚合
     * 限制：只能直接进行sum、max、min、count的聚合操作，无法直接计算平均值
     */
    val conf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName(this.getClass.getSimpleName.replace("$", ""))
    val sc = new SparkContext(conf)

    val wordsRDD: RDD[String] = sc.parallelize(List("a", "a", "a", "b", "b", "b"))

    val kvRDD: RDD[(String, Int)] = wordsRDD
      .map(w => (w, 1))

    kvRDD
      // 先对key进行分组，再对value进行聚合
      .reduceByKey((i1: Int, i2: Int) => {
        // sum聚合、count（KV格式的RDD的V必须是1）
        i1 + i2
        // max聚合
        //        Math.max(i1,i2)

        // min聚合
        //        Math.min(i1,i2)

        // 无法直接进行平均值的统计
      })
      .foreach(println)


    kvRDD
      // 先对key进行分组，再对value进行聚合
      // 简写
      .reduceByKey(_ + _)
      .foreach(println)

    val sRDD: RDD[(String, Int)] = sc.parallelize(List(("001", 20), ("001", 21), ("001", 23), ("002", 21), ("002", 23)))

    // 计算相同Key的所有Value的平均值
    // 第一种：groupByKey
    sRDD
      .groupByKey()
      .map(kv => {
        val id: String = kv._1
        s"$id,${kv._2.sum.toDouble / kv._2.size}"
      })
      .foreach(println)

    // 第二种：reduceByKey
    sRDD
      .map(t2 => (t2._1, (t2._2.toDouble, 1)))
      .reduceByKey((t2_1: (Double, Int), t2_2: (Double, Int)) => {
        (t2_1._1 + t2_2._1, t2_1._2 + t2_2._2)
      })
      .map(t2 => s"${t2._1},${t2._2._1 / t2._2._2}")
      .foreach(println)


    while (true) {

    }
  }

}
